package com.sunzm.spark.core.rdd.action

import java.sql.DriverManager

import com.alibaba.fastjson.JSON
import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SparkRDDAction {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RDD转换类算子示例")
      .setMaster("local[*]")
    val sc: SparkContext = new SparkContext(conf)

    val dataRDD: RDD[String] = sc.textFile("data/spark/rdd/order.txt")

    val filterRDD: RDD[String] = dataRDD.filter(line => {
      val jSONObject = JSON.parseObject(line)

      val orderId = jSONObject.getString("orderId")

      StringUtils.equals("o001", orderId)
    })

    //count: 返回RDD中元素的个数
    val count: Long = filterRDD.count()
    println(s"count： ${count}")

    //take(num: Int): 返回一个由RDD的前n个元素组成的数组
    val take3Array: Array[String] = dataRDD.take(3)
    //take3Array.mkString(",") 表示数组中的元素用逗号分隔，形成一个字符串
    println(s"take3Array： ${take3Array.mkString(",")}")

    //collect： 以数组Array的形式返回数据集的所有元素
    val collectArray: Array[String] = filterRDD.collect()
    println(s"collectArray： ${collectArray.mkString(",")}")

    // 保存成Text文件
    //dataRDD.saveAsTextFile("output")

    println("-----foreach---------")
    //foreach 没有返回值，需要对数据进行什么操作，直接在 foreach 的函数中进行
    dataRDD.foreach(line => {
      println(line)
    })

    //foreachPartition一次遍历一个分区
    dataRDD.foreachPartition(p => {
      //获取数据库连接, 如果在foreach中获取连接，有多少条数据就要获取几次数据库连接
      //而在foreachPartition中，有几个分区就获取几次连接
      //数据库连接是很宝贵的，应该尽量减少获取数据库连接的次数
      val conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/mydb", "root", "root")

      val sqlStr = "INSERT INTO t_order_tmp(orderId, orderTime) VALUES(?, ?)"

      val pstmt = conn.prepareStatement(sqlStr)

      //开始操作数据
      p.foreach(line => {

        val jSONObject = JSON.parseObject(line)

        val orderId = jSONObject.getString("orderId")
        val orderTime = jSONObject.getString("orderTime")

        pstmt.setString(1, orderId)
        pstmt.setString(2, orderTime)

        //执行插入操作
        //pstmt.execute()
      })

      //关闭连接
      if(pstmt != null){
        pstmt.close()
      }

      if(conn != null){
       conn.close()
      }
    })

    sc.stop()
  }
}
